﻿// Copyright (c) 2012 - 2020 Jared Zheng <jaredz at outlook dot com>
// Buddha Framework is licensed under Mulan PSL v2.
// You can use this software according to the terms and conditions of the Mulan PSL v2.
// You may obtain a copy of Mulan PSL v2 at:
//          http://license.coscl.org.cn/MulanPSL2
// THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
// EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
// MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
// See the Mulan PSL v2 for more details.
//
// Create : 2012-12-01 version 0.1 Jared Zheng <jaredz at outlook dot com>
// Update : 2018-12-05 version 3.1.5 Jared Zheng <jaredz at outlook dot com>

#ifndef __ROUTINE_IMP_H__
#define __ROUTINE_IMP_H__

#pragma once

#include "IGateRoutine.h"
#include "gatesession.h"
#include "routinedef.h"

///////////////////////////////////////////////////////////////////////////////////////////////////////////
// CCenterRoutine : 
class CGateRoutine : public IGateRoutine
{
    DECLARE_RTTI_CREATE( CGateRoutine )
public:
    CGateRoutine(void);
    virtual ~CGateRoutine(void);
    // CEventHandler
    virtual UInt OnHandle(Int nEvent, uintptr_t utData, ULLong ullParam) OVERRIDE;
    virtual UInt OnHandle(Int nEvent, CEventBase&, ULLong) OVERRIDE;
    virtual UInt OnHandle(Int nEvent, CStream& Stream, ULLong ullParam) OVERRIDE;
    virtual UInt OnHandle(Int nEvent, UInt uCount) OVERRIDE;
    // ICommonRoutine
    virtual bool Start(ICommonServer* pServer) OVERRIDE;
    virtual bool Pause(bool bPause = true) OVERRIDE;
    virtual bool Update(void) OVERRIDE;
    virtual void Stop(void) OVERRIDE;

    virtual UInt     GetThread(void) OVERRIDE; //
    virtual Socket_t GetSocket(UInt uType, UInt uId) OVERRIDE;//
    virtual bool     SetTimeout(UInt uTimeout = 0) OVERRIDE;

    virtual bool AddQueue(Int nEvent, uintptr_t utData = 0, ULLong ullParam = 0, QUEUE_TYPE eType = QUEUE_TYPE_NONE) OVERRIDE;
    virtual bool AddQueue(Int nEvent, CEventBase& EventRef, ULLong ullParam = 0, QUEUE_TYPE eType = QUEUE_TYPE_REFCOUNT) OVERRIDE;
    virtual bool AddQueue(Int nEvent, CStream& Stream, ULLong ullParam = 0) OVERRIDE;
private:
    bool  LoadExtConfig(bool bUpdate = false);
    bool  LoadExtBusiness(CKVNode& Root);
private:
    // 网关收到的在线业务功能RID
    bool  AddRoutineId(ULLong ullRoutineId);
    bool  RemoveRoutineId(ULLong ullRoutineId);
    bool  CheckRoutineId(ULLong ullRoutineId);

    bool  LinkRoutineId(ULLong ullRoutineId);
    bool  LinkAckRoutineId(ULLong ullRoutineId);

    bool  GetAllRoutineId(CGateSession::ARY_ROUTINEID& aRID);
    bool  RemoveAllRoutineId(void);

    bool  OnMasterVisa(CEventBase& EventRef, ULLong ullParam);

    bool  OnExternalBond(CEventBase& EventRef, ULLong ullParam);
    bool  OnExternalLogin(CEventBase& EventRef, ULLong ullParam);

    bool  OnRoutine(CEventBase& EventRef, ULLong ullParam);
    bool  OnRoutineRid(ULLong ullParam, CNETRoutine* pRoutine);
    bool  OnRoutineAck(CEventBase& EventRef, ULLong ullParam);
    //-------------------------------------------------------------
    bool  CheckQueueSessionStatus(void);
    bool  CheckConfirmSessionStatus(void);
    //-------------------------------------------------------------
    bool  GetQueueSession(ULLong ullParam, CGateSessionPtr& SessionPtr);
    bool  RemoveQueueSession(ULLong ullParam);

    bool  GetSession(ULLong ullParam, CGateSessionPtr& SessionPtr);
    bool  AddSession(ULLong ullParam, ULLong ullAuthor, CGateSessionPtr& SessionPtr);
    bool  RemoveSession(ULLong ullParam, uintptr_t& utLiveData, CGateSessionPtr& SessionPtr);

    bool  LinkSessionRId(ULLong ullParam, CNETRoutine* pRoutine);
    bool  LinkAckSessionRId(ULLong ullParam, ULLong ullRId);
    bool  UnlinkSessionRId(ULLong ullRId = 0);

    LLong IncrementUpdate(LLong llUpdate, ULLong ullParam); // / 授权网关有一样的功能
    //-------------------------------------------------------------
    UACK  CheckSessionTime(LLong llUpdate, ULLong ullParam, UInt uCurStatus, UInt uSetStatus, CGateSessionPtr& SessionPtr);
    //-------------------------------------------------------------
    bool  ResetAck(CPAKHeadBase* pPAKHead, UEVENT uEvent, UACK uAck = (UACK)ERRORD_OKAY, UTYPE uType = (UTYPE)PAKT_GATE);
private:
    struct tagCONFIRM_TIME : public MObject
    {
        LLong    llTimeout;
        ULLong   ullAuthor;
    };
    typedef tagCONFIRM_TIME CONFIRM_TIME, *PCONFIRM_TIME;

    struct tagROUTINEID_INFO : public MObject
    {
        LLong    llReqCount;
        LLong    llAckCount;
        UInt     uStatus;
        UInt     uUpdate;
    };
    typedef tagROUTINEID_INFO ROUTINEID_INFO, *PROUTINEID_INFO;

    typedef CTMap<ULLong, CONFIRM_TIME>           CONFIRM_MAP, *PCONFIRM_MAP;
    typedef CTMap<ULLong, CONFIRM_TIME>::PAIR     CONFIRM_MAP_PAIR;

    typedef CTMap<ULLong, ROUTINEID_INFO>         RID_INFO_MAP, *PROUTINEID_INFO_MAP;
    typedef CTMap<ULLong, ROUTINEID_INFO>::PAIR   RID_INFO_MAP_PAIR;
    typedef tagSERVER_CHAIN<RID_INFO_MAP>         RID_INFO_MAP_CHAIN;
private:
    ICommonServer*        m_pServer;         ///< 框架例程处理对应的服务器
    PULLong               m_pKeyPtr;
    time_t                m_tExtConfig;      ///< 扩展配置文件修改时间
    Int                   m_nTime;           ///< update()倍数次进行配置文件修改检测
    Int                   m_nUpdate;         ///< 
    UInt                  m_uThreads;        ///< [1, cores]
    UInt                  m_uTimeout;        ///< 异步事件队列工作线程超时检测的时间, 单位[S];
    //-------------------------------------------------------------
    UInt                  m_uMasterId;
    UInt                  m_uTimeoutTick;
    UInt                  m_uTimeoutMin;
    UInt                  m_uTimeoutMax;
    bool                  m_bCompleteLive;
    bool                  m_bRID2External;
    // 在线业务功能服务RID
    RID_INFO_MAP_CHAIN    m_RIdInfo;

    GATE_SESSION_MAP      m_QueueMap;
    CSyncLock             m_QueueLock;

    CONFIRM_MAP           m_ConfirmMap;
    GATE_SESSION_MAP      m_SessionMap;
    CSyncLock             m_SessionLock;
    //-------------------------------------------------------------
    CEventQueuePtr        m_EventQueuePtr;   ///< 事件队列, 每工作线程使用专属的redis连接
    CCommonRoutinePtr     m_BusinessPtr;     ///< 和产品业务相关
};

INLINE CGateRoutine::CGateRoutine(void)
: m_pServer(nullptr)
, m_pKeyPtr(nullptr)
, m_tExtConfig(0)
, m_nTime(DATAD_UPDATE)
, m_nUpdate(0)
, m_uThreads(0)
, m_uTimeout(0)
, m_uMasterId(0)
, m_uTimeoutTick(SESSION_TIME_OUT_TICK)
, m_uTimeoutMin(SESSION_TIME_OUT_MIN)
, m_uTimeoutMax(SESSION_TIME_OUT_MAX)
, m_bCompleteLive(false)
, m_bRID2External(false)
{
    DEV_DEBUG(TF(" + %s new instance at %p"), GetRTTI().GetName(), this);
}

INLINE CGateRoutine::~CGateRoutine(void)
{
    DEV_DEBUG(TF(" - %s delete instance at %p"), GetRTTI().GetName(), this);
}

INLINE UInt CGateRoutine::GetThread(void)
{
    return 0;
}

INLINE Socket_t CGateRoutine::GetSocket(UInt, UInt)
{
    return 0;
}

INLINE bool CGateRoutine::GetQueueSession(ULLong ullParam, CGateSessionPtr& SessionPtr)
{
    CSyncLockWaitScope scope(m_QueueLock);
    PINDEX index = m_QueueMap.FindIndex(ullParam);
    if (index != nullptr)
    {
        SessionPtr = m_QueueMap[index];
    }
    return (SessionPtr != nullptr);
}

INLINE bool CGateRoutine::RemoveQueueSession(ULLong ullParam)
{
    CSyncLockWaitScope scope(m_QueueLock);
    return m_QueueMap.Remove(ullParam);
}

INLINE bool CGateRoutine::GetSession(ULLong ullParam, CGateSessionPtr& SessionPtr)
{
    CSyncLockWaitScope scope(m_SessionLock);
    PINDEX index = m_SessionMap.FindIndex(ullParam);
    if (index != nullptr)
    {
        SessionPtr = m_SessionMap[index];
    }
    return (SessionPtr != nullptr);
}

INLINE bool CGateRoutine::AddSession(ULLong ullParam, ULLong ullAuthor, CGateSessionPtr& SessionPtr)
{
    CSyncLockWaitScope scope(m_SessionLock);
    assert(m_ConfirmMap.Find(ullParam) == nullptr);
    assert(m_SessionMap.Find(ullParam) == nullptr);

    CONFIRM_TIME ct;
    ct.ullAuthor = ullAuthor;
    ct.llTimeout = SessionPtr->GetUpdate() + m_uTimeoutMax * SESSION_TIME_COMPLETE;
    m_ConfirmMap.Add(ullParam, ct);
    m_SessionMap.Add(ullParam, SessionPtr);
    return true;
}

INLINE bool CGateRoutine::RemoveSession(ULLong ullParam, uintptr_t& utLiveData, CGateSessionPtr& SessionPtr)
{
    if (utLiveData != PAKT_EXTERNAL_READY)
    {
        CSyncLockWaitScope scope(m_SessionLock);
        PINDEX index = m_SessionMap.FindIndex(ullParam);
        if (index != nullptr)
        {
            SessionPtr = m_SessionMap[index];
            m_SessionMap.RemoveAt(index);
        }
        // 业务区流程完成, 但是授权和外网的连接还没有断开的
        // 如果当前业务区有异常导致连接断开, 通知授权重新开始业务区签证流程
        utLiveData = 0;
        CONFIRM_MAP_PAIR* pPair = m_ConfirmMap.Find(ullParam);
        if (pPair != nullptr)
        {
            utLiveData = (uintptr_t)pPair->m_V.ullAuthor;
            m_ConfirmMap.RemoveAt(reinterpret_cast<PINDEX>(pPair));
        }
    }
    return (SessionPtr != nullptr);
}

INLINE bool CGateRoutine::LinkSessionRId(ULLong ullParam, CNETRoutine* pRoutine)
{
    CGateSessionPtr SessionPtr;
    if (GetSession(ullParam, SessionPtr))
    {
        size_t stSize = sizeof(SESSION_ID) + SessionPtr->GetData().Size();
        // 确保缓存容量能够传递所有数据
        if (pRoutine->Cache(true) < stSize)
        {
            pRoutine->Alloc(m_pServer->GetNetworkPtr(), true);
            assert(pRoutine->Cache(true) >= stSize);
        }
        CStream& Stream = pRoutine->WriteStream(m_pServer->GetNetworkPtr());
        Stream.Write(SessionPtr->GetSession(), sizeof(SESSION_ID));

        CBufReadStream& StreamData = SessionPtr->GetData();
        if (StreamData.Size() > 0)
        {
            Stream.Write(StreamData.GetBuf(), StreamData.Size());
#ifdef __DUMP_STEP_DETAIL__
            LOGV_DUMP(m_pServer->GetFileLog(), TF("[业务网关例程]网关会话[%p]业务连接请求, 对应业务RID=%llX, 增加SESSION_ID数据+自定义数据大小=%d!"), ullParam, pRoutine->GetRoutineId(), StreamData.Size());
#endif
        }
        SessionPtr->Link(pRoutine->GetRoutineId());
        LinkRoutineId(pRoutine->GetRoutineId());
        return true;
    }
    else
    {
#ifdef __DUMP_STEP_DETAIL__
        LOGV_WARN(m_pServer->GetFileLog(), TF("[业务网关例程]网关会话[%p]业务请求失败, 无法通过连接找到会话数据!"), ullParam);
#endif
        CPAKSrvFaultAck Ack((UACK)ROUTINE_ERROR_VOIDNESS_SESSION);
        Ack.SetSessionId(ullParam);
        m_pServer->GetNetworkPtr()->Send((Socket_t)ullParam, Ack, CNETTraits::SEND_CLOSE);
        return false;
    }
}

INLINE bool CGateRoutine::LinkAckSessionRId(ULLong ullParam, ULLong ullRId)
{
    CGateSessionPtr SessionPtr;
    if (GetSession(ullParam, SessionPtr))
    {
        SessionPtr->LinkAck(ullRId);
        LinkAckRoutineId(ullRId);
//#ifdef __DUMP_STEP_DETAIL__
//        LOGV_DUMP(m_pServer->GetFileLog(), TF("[业务网关例程]网关会话[%p]业务连接请求返回成功, 对应业务RID=%llX!"), ullParam, ullRId);
//#endif
        return true;
    }
    return false;
}

INLINE bool CGateRoutine::UnlinkSessionRId(ULLong ullRId)
{
    CSyncLockWaitScope scope(m_SessionLock);
    for (PINDEX index = m_SessionMap.GetFirst(); index != nullptr; )
    {
        GATE_SESSION_MAP_PAIR* pPair = m_SessionMap.GetNext(index);
        pPair->m_V->Unlink(ullRId);
    }
    return true;
}

INLINE LLong CGateRoutine::IncrementUpdate(LLong llUpdate, ULLong ullParam)
{
    ullParam  = *(m_pKeyPtr + ullParam % 2047);
    ullParam += *(m_pKeyPtr + llUpdate % 2047);
    return (llUpdate + 200 + (LLong)(ullParam & 0x03FF));
}

INLINE UACK CGateRoutine::CheckSessionTime(LLong llUpdate, ULLong ullParam, UInt uCurStatus, UInt uSetStatus, CGateSessionPtr& SessionPtr)
{
    if (m_uMasterId == 0)
    {
#ifdef __DUMP_STEP_DETAIL__
        DEV_DUMP(TF("[业务网关例程]%p检测连接时和业务管理连接已经断开"), ullParam);
#endif
        return (UACK)ROUTINE_ERROR_REQ_MASTER;
    }
    if (GetQueueSession(ullParam, SessionPtr) == false)
    {
#ifdef __DUMP_STEP_DETAIL__
        DEV_DUMP(TF("[业务网关例程]%p检测连接和会话映射关系失败2"), ullParam);
#endif
        return (UACK)ROUTINE_ERROR_VOIDNESS_DATA;
    }
    if (SessionPtr->TrySetUpdate(llUpdate, IncrementUpdate(llUpdate, ullParam)) == false)
    {
#ifdef __DUMP_STEP_DETAIL__
        DEV_DUMP(TF("[业务网关例程]%p检测连接的会话2时间戳失败%llX---%llX"), ullParam, SessionPtr->GetUpdate(), llUpdate);
#endif
        return (UACK)ROUTINE_ERROR_VOIDNESS_STATUS;
    }
    if (SessionPtr->TrySetStatus(uCurStatus, uSetStatus) == false)
    {
#ifdef __DUMP_STEP_DETAIL__
        DEV_DUMP(TF("[业务网关例程]%p检测连接的会话2状态失败%d---%d"), ullParam, SessionPtr->GetStatus(), uCurStatus);
#endif
        return (UACK)ROUTINE_ERROR_VOIDNESS_STATUS;
    }
    return (UACK)ERRORD_OKAY;
}

INLINE bool CGateRoutine::ResetAck(CPAKHeadBase* pPAKHead, UEVENT uEvent, UACK uAck, UTYPE uType)
{
    pPAKHead->ResetHead(uEvent, uType, uAck);
    return true;
}

#endif // __ROUTINE_IMP_H__
